package com.gitee.xhs.customWaterMark;

import com.gitee.xhs.EnvironmentConfig;
import com.gitee.xhs.FlinkEnvironmentFactory;
import com.gitee.xhs.WatermarkRecordMakerProcessFunction;
import com.gitee.xhs.WatermarkSourceFactory;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static com.gitee.xhs.CustomDefaultTimestampAssigner.*;
import static com.gitee.xhs.CustomDefaultWatermarkGenerator.CUSTOM_DEFAULT_WATERMARK_GENERATOR;
import static com.gitee.xhs.customWaterMark.CustomWaterStrategy.*;

/**
 * @author BigHead
 * @date 2022/7/14 16:24
 * @description CustomWaterMarkMain
 * <p>
 * 执行ProcessOperator算子的时候
 * 首先执行 processElement 再执行 processWatermark 这种执行顺序会导致 ProcessOperator 中的
 * ctx.timerService().currentWatermark() 会滞后当前数据的时间戳
 */
public class CustomWaterMarkMain {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamExecutionEnvironment = FlinkEnvironmentFactory.webUi(new EnvironmentConfig(null));

        streamExecutionEnvironment
                .addSource(
                        WatermarkSourceFactory.defaultTransactionBoundOneSecondSource()
                )
                .assignTimestampsAndWatermarks(
                        new CustomWaterStrategy()
                )
                .process(
                        new WatermarkRecordMakerProcessFunction(CUSTOM_WATER_STRATEGY, CUSTOM_DEFAULT_WATERMARK_GENERATOR, CUSTOM_DEFAULT_WATERMARK_ASSIGNER)
                )
                .print();


        streamExecutionEnvironment.execute("CustomWaterMarkMain");
    }

}
